package tailfile

import (
	"context"
	"github.com/Shopify/sarama"
	"github.com/hpcloud/tail"
	"github.com/sirupsen/logrus"
	"logAgent/kafkafile"
	"strings"
	"time"
)

type tailTask struct {
	path string
	topic string
	tObj *tail.Tail
	ctx context.Context
	cancel context.CancelFunc
}

var (
	TailObj *tail.Tail
)


func NewTailTask(path, topic string)*tailTask {
	ctx, cancel := context.WithCancel(context.Background())

	tt := & tailTask{
		path: path,
		topic: topic,
		tObj: TailObj,
		ctx: ctx,
		cancel: cancel,
	}
	return tt
}

func (t *tailTask) init()(err error) {
	config := tail.Config{
		ReOpen: true,
		Follow: true,
		Location: &tail.SeekInfo{Offset: 0, Whence: 0},
		MustExist: false,
		Poll: true,
	}
	//打开文件读取数据
	t.tObj, err = tail.TailFile(t.path, config)
		if err != nil {
			logrus.Error("create tailObj for path： %是failed,err:%v\n", TailObj, err)
			return
		}
	return
}


func(t *tailTask) run() {
	for {
		select {
		case <- t.ctx.Done():   // 只要调用t.cancel() 就会收到信号，Done()
			logrus.Infof("path:%s is stopping...", t.path)
			return
		//循环读取数据
		case line, ok := <-t.tObj.Lines:
			if !ok {
				logrus.Warn("tail file close reopen, path:%s\n", t.path)
				time.Sleep(time.Second) // 读取出错等一秒
				continue
			}
			//如果是空行就略过
			if len(strings.Trim(line.Text,"\r")) == 0{
				logrus.Info("出现空行拉,直接跳过...")
				continue
			}
			//利用通道可以将代码改成异步的
			//把读出来的一行包装成kafka里面的msg类型
			msg := &sarama.ProducerMessage{}
			msg.Topic = t.topic
			msg.Value = sarama.StringEncoder(line.Text)
			kafkafile.ToMsgChan(msg)
		}
	}
}



